feat: implement asynchronous audit logging with RabbitMQ and DLQ#100
Conversation
|
@N-thnI Great news! 🎉 Based on an automated assessment of this PR, the linked Wave issue(s) no longer count against your application limits. You can now already apply to more issues while waiting for a review of this PR. Keep up the great work! 🚀 |
|
Caution Review failedThe pull request is closed. ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (3)
📝 WalkthroughWalkthroughIntroduces RabbitMQ-backed audit pipeline: publishes audit events to an Changes
Sequence Diagram(s)sequenceDiagram
participant App as Application
participant AS as AuditService
participant RMQ as RabbitMQ
participant AC as AuditConsumer
participant DB as Database
participant DLQ as AUDIT_LOGS_DLQ
participant Fallback as Fallback Handler
rect rgba(76, 175, 80, 0.5)
note over App,DB: Happy path
App->>AS: logAudit(entry)
AS->>RMQ: sendToQueue(AUDIT_LOGS, payload, persistent)
RMQ-->>AS: true
AS-->>App: return
RMQ->>AC: deliver(message)
AC->>DB: prisma.auditTrail.create(parsed)
DB-->>AC: success
AC->>RMQ: ack(message)
end
rect rgba(255, 152, 0, 0.5)
note over App,Fallback: Publish failure -> fallback
App->>AS: logAudit(entry)
AS->>RMQ: sendToQueue(AUDIT_LOGS, payload, persistent)
RMQ-->>AS: false
AS->>Fallback: handleAuditFailure(entry + error)
Fallback->>Fallback: append lost-audits.log
Fallback->>Fallback: sendEmail(if configured)
AS-->>App: return
end
rect rgba(244, 67, 54, 0.5)
note over AC,DLQ: Consumer retries -> DLQ
RMQ->>AC: deliver(message)
AC->>AC: retry loop (exponential backoff)
AC->>DB: prisma.auditTrail.create()
DB-->>AC: error (after retries)
AC->>RMQ: nack(message, false, false) -> routed to DLQ
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/index.ts`:
- Around line 108-110: The call to startAuditConsumer inside startServer can
throw (e.g., from getRabbitMQChannel/queue assertions) and currently lives in
the outer try/catch, causing the whole service to exit on audit-queue failure;
change this so audit consumer startup is isolated: move the dynamic
import("./jobs/auditConsumer")/startAuditConsumer invocation out of the main
startServer critical path and wrap it in its own try/catch (or start it without
awaiting) so any exception from startAuditConsumer or getRabbitMQChannel is
caught, logged (including error details) and the server continues running
without queue-based features.
In `@src/jobs/auditConsumer.ts`:
- Around line 17-18: The handler currently calls JSON.parse on msg.content
outside the retry/DLQ try/catch so malformed payloads escape ack/nack handling;
move the parsing of msg.content (the lines that create content and entry) inside
the existing try/catch used for processing, and in the catch ensure you call
msg.nack() (or msg.ack() as appropriate for fatal vs retryable) so malformed
messages are nacked and follow the retry/DLQ path; update the same pattern for
the other block referenced (lines 39-55) so every JSON.parse is wrapped and
leads to msg.nack() on parse errors.
In `@src/services/audit/auditService.ts`:
- Around line 65-98: The current try block that writes the fallback file also
contains the sendEmail call, so if fs.appendFileSync or mkdirSync fails the
alert email is skipped; refactor the block in auditService.ts so that
sendEmail(config.notification.alertEmail, subject, body) is executed outside the
filesystem try (or placed into a finally) — keep building subject/body using
entry and error as shown, wrap the sendEmail call in its own catch to log
failures via logger.error, and ensure fallbackPath/logger messages remain in the
file-write try while email alerting is always attempted regardless of filesystem
errors.
- Around line 59-63: The logger.error call (and the alert email path) currently
embeds the full audit entry object; create and use a sanitizer (e.g.
redactSensitiveFields or sanitizeAuditEntry) that returns a copy of entry with
sensitive keys removed or replaced (e.g. strip or mask oldValue/newValue/any
payload fields) and pass that sanitized object to logger.error and to the
email-sending routine (the alert email body path). Update the code paths that
reference entry (the logger.error call and the alert email construction) to use
the sanitized version instead of the raw entry to avoid leaking sensitive audit
payloads.
- Around line 30-45: The current use of channel.sendToQueue(QUEUES.AUDIT_LOGS,
...) relies on the write-buffer boolean and must be replaced with publisher
confirms: create a ConfirmChannel via connection.createConfirmChannel() and
publish using ConfirmChannel.sendToQueue with the callback form sendToQueue(...,
{ persistent: true, mandatory: true }, (err, ok) => { ... }) so you only treat
the message as delivered on callback success (err === null); also attach a
channel 'return' event handler to detect unroutable messages and handle/alert
accordingly; update any logic in the audit publishing path that currently throws
on a boolean false to instead wait for the confirm callback and handle
err/return events.
In `@tests/simple.test.ts`:
- Around line 1-5: The current test only asserts a trivial value and doesn't
exercise the audit logging system; replace the trivial spec with
unit/integration tests that exercise the real code paths: call the
AuditPublisher.publishAuditEvent (or equivalent) to assert it enqueues to
RabbitMQClient.publish, simulate publish failures to verify the retry logic on
publishRetry, assert messages routed to DLQHandler.handleDeadLetter after
exceeded retries, verify FileFallback.persist (or fallbackToFile) is invoked
when RabbitMQ is unavailable, and mock sendCriticalAlert (or
AlertService.notifyCritical) to confirm alerting on persistent failures; use
spies/mocks for RabbitMQClient, DLQHandler, FileFallback, and AlertService to
assert behavior and add an integration-style test that runs the async flow
end-to-end with a test RabbitMQ or in-memory emulator.
- Around line 1-2: The test file uses single quotes which violate the project's
Prettier rules; update string delimiters in the test by replacing single quotes
with double quotes in the describe and it calls (e.g., the "describe" and "it"
string literals) so the file uses double-quoted strings consistently per the
project's Prettier configuration.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 7c84ffff-51ef-457e-9479-5ed01ec9c20f
📒 Files selected for processing (8)
src/config/env.tssrc/config/rabbitmq.tssrc/index.tssrc/jobs/auditConsumer.test.tssrc/jobs/auditConsumer.tssrc/services/audit/auditService.test.tssrc/services/audit/auditService.tstests/simple.test.ts
| // Start audit consumer (AUDIT_LOGS → database) | ||
| const { startAuditConsumer } = await import("./jobs/auditConsumer"); | ||
| await startAuditConsumer(); |
There was a problem hiding this comment.
Isolate audit-consumer startup failure from API startup.
This await sits inside the outer startServer() try/catch, so any getRabbitMQChannel()/queue-assertion failure from startAuditConsumer() now takes the whole service down at Lines 188-190. That contradicts the earlier “continue without queue-based features” behavior and turns an audit-queue misconfiguration into a full API outage.
Suggested fix
- const { startAuditConsumer } = await import("./jobs/auditConsumer");
- await startAuditConsumer();
+ try {
+ const { startAuditConsumer } = await import("./jobs/auditConsumer");
+ await startAuditConsumer();
+ } catch (error) {
+ logger.error(
+ "Audit consumer failed to start; continuing with fallback-only audit persistence.",
+ error,
+ );
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| // Start audit consumer (AUDIT_LOGS → database) | |
| const { startAuditConsumer } = await import("./jobs/auditConsumer"); | |
| await startAuditConsumer(); | |
| // Start audit consumer (AUDIT_LOGS → database) | |
| try { | |
| const { startAuditConsumer } = await import("./jobs/auditConsumer"); | |
| await startAuditConsumer(); | |
| } catch (error) { | |
| logger.error( | |
| "Audit consumer failed to start; continuing with fallback-only audit persistence.", | |
| error, | |
| ); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/index.ts` around lines 108 - 110, The call to startAuditConsumer inside
startServer can throw (e.g., from getRabbitMQChannel/queue assertions) and
currently lives in the outer try/catch, causing the whole service to exit on
audit-queue failure; change this so audit consumer startup is isolated: move the
dynamic import("./jobs/auditConsumer")/startAuditConsumer invocation out of the
main startServer critical path and wrap it in its own try/catch (or start it
without awaiting) so any exception from startAuditConsumer or getRabbitMQChannel
is caught, logged (including error details) and the server continues running
without queue-based features.
| const content = msg.content.toString(); | ||
| const entry = JSON.parse(content); |
There was a problem hiding this comment.
Nack malformed messages instead of bypassing the DLQ path.
JSON.parse() happens before the retry/DLQ try/catch, so a bad payload exits the handler before either ack() or nack() runs. That leaves poison messages outside the intended DLQ flow.
Suggested fix
- const content = msg.content.toString();
- const entry = JSON.parse(content);
+ let entry: any;
+ try {
+ entry = JSON.parse(msg.content.toString());
+ } catch (error: any) {
+ logger.error("Invalid audit payload, moving message to DLQ", {
+ error: error.message || error,
+ });
+ channel.nack(msg, false, false);
+ return;
+ }Also applies to: 39-55
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/jobs/auditConsumer.ts` around lines 17 - 18, The handler currently calls
JSON.parse on msg.content outside the retry/DLQ try/catch so malformed payloads
escape ack/nack handling; move the parsing of msg.content (the lines that create
content and entry) inside the existing try/catch used for processing, and in the
catch ensure you call msg.nack() (or msg.ack() as appropriate for fatal vs
retryable) so malformed messages are nacked and follow the retry/DLQ path;
update the same pattern for the other block referenced (lines 39-55) so every
JSON.parse is wrapped and leads to msg.nack() on parse errors.
| const success = channel.sendToQueue( | ||
| QUEUES.AUDIT_LOGS, | ||
| Buffer.from(JSON.stringify(payload)), | ||
| { persistent: true }, | ||
| ); | ||
|
|
||
| if (success) { | ||
| logger.debug("Audit entry published to queue", { | ||
| eventType: entry.eventType, | ||
| entityType: entry.entityType ?? undefined, | ||
| entityId: entry.entityId ?? undefined, | ||
| action: entry.action, | ||
| oldValue: entry.oldValue ?? undefined, | ||
| newValue: entry.newValue ?? undefined, | ||
| performedBy: entry.performedBy ?? undefined, | ||
| }, | ||
| }); | ||
| return; | ||
| } | ||
|
|
||
| // If queue is full or other internal issue, treat as failure | ||
| throw new Error("RabbitMQ sendToQueue returned false"); |
There was a problem hiding this comment.
❓ Verification inconclusive
Script executed:
cat -n src/services/audit/auditService.ts | sed -n '20,50p'Repository: Pi-Defi-world/acbu-backend
Repository: Pi-Defi-world/acbu-backend
Exit code: 0
stdout:
20 * Falls back to local file and alerting if RabbitMQ is unavailable.
21 */
22 export async function logAudit(entry: AuditEntry): Promise<void> {
23 try {
24 const channel = getRabbitMQChannel();
25 const payload = {
26 ...entry,
27 timestamp: new Date().toISOString(),
28 };
29
30 const success = channel.sendToQueue(
31 QUEUES.AUDIT_LOGS,
32 Buffer.from(JSON.stringify(payload)),
33 { persistent: true },
34 );
35
36 if (success) {
37 logger.debug("Audit entry published to queue", {
38 eventType: entry.eventType,
39 action: entry.action,
40 });
41 return;
42 }
43
44 // If queue is full or other internal issue, treat as failure
45 throw new Error("RabbitMQ sendToQueue returned false");
46 } catch (error: any) {
47 logger.warn("RabbitMQ audit publish failed, using fallback", {
48 error: error.message || error,
49 eventType: entry.eventType,
50 });
amqplib sendToQueue return value boolean backpressure vs delivery
In amqplib, the channel.sendToQueue(queue, content, [options], [callback]) method returns a boolean value indicating local flow control (backpressure), not delivery success to the broker. - true: The message was accepted into the channel's output buffer; you can continue sending immediately. - false: The channel's write buffer is full (backpressure). Wait for the channel's 'drain' event before sending more messages to avoid memory exhaustion. This boolean is unrelated to whether the message reaches the queue or broker. Even if false is returned, the message is buffered internally by amqplib and will be sent when possible (assuming no crash). Do not resend the message; just pause publishing until 'drain'. For actual delivery confirmation (broker acknowledgment that the message was processed/stored): - Use a ConfirmChannel (from connection.createConfirmChannel). - Pass a callback as the last argument: ch.sendToQueue(..., function(err, ok) { ... }). - err: Non-null if delivery failed (e.g., unroutable with mandatory: true). - ok: true if successfully delivered to broker. - Or use ch.waitForConfirms after sending (for all pending messages). - Listen for 'return' events on unroutable mandatory messages. sendToQueue bypasses exchanges (direct to queue), so no routing issues unless queue doesn't exist (use mandatory: true and 'return' handler to detect). Examples: // Regular channel - only backpressure boolean let ok = ch.sendToQueue('q', Buffer.from('msg')); if (!ok) { // Wait for 'drain' await new Promise(resolve => ch.once('drain', resolve)); } // ConfirmChannel - delivery callback ch.sendToQueue('q', Buffer.from('msg'), {}, (err, ok) => { if (err) console.error('Delivery failed'); else console.log('Delivered'); });
Citations:
- 1: https://amqp-node.github.io/amqplib/channel_api.html
- 2: Confirm channel callback when publish/sendToQueue returns false amqp-node/amqplib#636
- 3: ConfirmChannel#publish / ConfirmChannel#sendToQueue should return a promise ? amqp-node/amqplib#89
- 4: Publish returning false: should I send the message again or not? amqp-node/amqplib#713
- 5: https://github.com/amqp-node/amqplib/blob/main/examples/tutorials/send.js
amqplib ConfirmChannel publisher confirms vs Channel sendToQueue
In amqplib, Channel.sendToQueue and ConfirmChannel publisher confirms serve different purposes for reliable message publishing to RabbitMQ. Channel.sendToQueue (on a regular Channel): - Publishes directly to a specific queue (bypassing exchange routing). - Returns boolean: true (keep sending) or false (wait for 'drain' event due to buffer full). - No broker confirmation of receipt/persistence - fire-and-forget, message can be lost on network failure or broker issues. - Behaves like stream.Writable for flow control. ConfirmChannel (created via connection.createConfirmChannel): - Enables RabbitMQ publisher confirms (broker ACK/NACK after message handled). - sendToQueue and publish accept optional callback(err, ok): called on broker confirm (ACK=ok, NACK=err). - Also has waitForConfirms: waits for all outstanding publishes since last call (or channel open) to be confirmed. - Provides at-least-once delivery guarantee when used with persistent messages and durable queues. Key differences: - Regular Channel: Local flow control only, no broker ACK. - ConfirmChannel: Adds broker-level confirmation for data safety. Usage example (Node.js): // Regular channel - no confirms const channel = await conn.createChannel; channel.sendToQueue('queue', Buffer.from('msg')); // Confirm channel - with confirms const confirmChannel = await conn.createConfirmChannel; confirmChannel.sendToQueue('queue', Buffer.from('msg'), {}, (err, ok) => { if (err) console.error('NACK'); else console.log('ACK'); }); // Or wait for batch confirmChannel.sendToQueue('queue', Buffer.from('msg')); await confirmChannel.waitForConfirms; For production reliability, use ConfirmChannel with persistent: true option.
Citations:
- 1: Confirm channel callback when publish/sendToQueue returns false amqp-node/amqplib#636
- 2: https://amqp-node.github.io/amqplib/channel_api.html
- 3: Publisher confirms the right way? amqp-node/amqplib#434
- 4: https://oneuptime.com/blog/post/2026-01-24-rabbitmq-publisher-confirms/view
- 5: https://www.rabbitmq.com/publishers.html
- 6: https://typestrong.org/typedoc-auto-docs/_types_amqplib/interfaces/ConfirmChannel.html
- 7: How to know if message is published ? amqp-node/amqplib#751
- 8: https://oneuptime.com/blog/post/2026-01-26-rabbitmq-publisher-confirms/view
Script executed:
# Check if ConfirmChannel is used anywhere in the codebase
rg "ConfirmChannel" --type typescriptRepository: Pi-Defi-world/acbu-backend
Repository: Pi-Defi-world/acbu-backend
Exit code: 2
stderr:
unrecognized file type: typescript
Script executed:
# Check RabbitMQ channel initialization and confirm settings
cat -n src/config/rabbitmq.tsRepository: Pi-Defi-world/acbu-backend
Repository: Pi-Defi-world/acbu-backend
Exit code: 0
stdout:
1 import amqp, { Channel, ChannelModel } from "amqplib";
2 import { config } from "./env";
3 import { logger } from "./logger";
4
5 let connection: ChannelModel | null = null;
6 let channel: Channel | null = null;
7
8 export async function connectRabbitMQ(): Promise<Channel> {
9 if (channel) {
10 return channel;
11 }
12
13 try {
14 connection = await amqp.connect(config.rabbitmqUrl);
15 const ch = await connection.createChannel();
16 channel = ch;
17 logger.info("RabbitMQ connected successfully");
18
19 await ch.assertExchange(EXCHANGES.DLX, "direct", { durable: true });
20 logger.info("Dead letter exchange asserted");
21
22 // Handle connection errors
23 connection.on("error", (err: Error) => {
24 logger.error("RabbitMQ connection error", err);
25 });
26
27 connection.on("close", () => {
28 logger.warn("RabbitMQ connection closed");
29 connection = null;
30 channel = null;
31 });
32
33 return ch;
34 } catch (error) {
35 logger.error("Failed to connect to RabbitMQ", error);
36 throw error;
37 }
38 }
39
40 export interface QueueOptions extends amqp.Options.AssertQueue {
41 deadLetterQueue?: string;
42 }
43
44 export async function assertQueueWithDLQ(
45 queueName: string,
46 options: QueueOptions = {},
47 ): Promise<amqp.Replies.AssertQueue> {
48 const ch = getRabbitMQChannel();
49 const dlqName = `${queueName}_dlq`;
50
51 await ch.assertQueue(dlqName, { durable: true });
52
53 await ch.bindQueue(dlqName, EXCHANGES.DLX, queueName);
54
55 const queueOptions: amqp.Options.AssertQueue = {
56 ...options,
57 durable: true,
58 arguments: {
59 ...options.arguments,
60 deadLetterExchange: EXCHANGES.DLX,
61 deadLetterRoutingKey: queueName,
62 },
63 };
64
65 return ch.assertQueue(queueName, queueOptions);
66 }
67
68 export async function disconnectRabbitMQ(): Promise<void> {
69 if (channel) {
70 await channel.close();
71 channel = null;
72 }
73 if (connection) {
74 await connection.close();
75 connection = null;
76 logger.info("RabbitMQ disconnected");
77 }
78 }
79
80 export function getRabbitMQChannel(): Channel {
81 if (!channel) {
82 throw new Error("RabbitMQ not connected. Call connectRabbitMQ() first.");
83 }
84 return channel;
85 }
86
87 // Queue names
88 export const QUEUES = {
89 USDC_CONVERSION: "usdc_conversion",
90 USDC_CONVERSION_DLQ: "usdc_conversion_dlq",
91 WITHDRAWAL_PROCESSING: "withdrawal_processing",
92 WITHDRAWAL_PROCESSING_DLQ: "withdrawal_processing_dlq",
93 REBALANCING: "rebalancing",
94 REBALANCING_DLQ: "rebalancing_dlq",
95 NOTIFICATIONS: "notifications",
96 NOTIFICATIONS_DLQ: "notifications_dlq",
97 OTP_SEND: "otp_send", // OTP delivery (email/SMS) via worker
98 OTP_SEND_DLQ: "otp_send_dlq",
99 WEBHOOKS: "webhooks",
100 WEBHOOKS_DLQ: "webhooks_dlq", // failed webhook deliveries for retry
101 KYC_PROCESSING: "kyc_processing",
102 KYC_PROCESSING_DLQ: "kyc_processing_dlq",
103 WALLET_ACTIVATION: "wallet_activation", // send XLM to user wallet when KYC fee paid
104 WALLET_ACTIVATION_DLQ: "wallet_activation_dlq",
105 ACBU_SAVINGS_VAULT_EVENTS: "acbu_savings_vault_events",
106 ACBU_SAVINGS_VAULT_EVENTS_DLQ: "acbu_savings_vault_events_dlq",
107 ACBU_LENDING_POOL_EVENTS: "acbu_lending_pool_events",
108 ACBU_LENDING_POOL_EVENTS_DLQ: "acbu_lending_pool_events_dlq",
109 ACBU_ESCROW_EVENTS: "acbu_escrow_events",
110 ACBU_ESCROW_EVENTS_DLQ: "acbu_escrow_events_dlq",
111 XLM_TO_ACBU: "xlm_to_acbu", // XLM deposit: sell XLM and mint ACBU to user
112 XLM_TO_ACBU_DLQ: "xlm_to_acbu_dlq",
113 USDC_CONVERT_AND_MINT: "usdc_convert_and_mint", // USDC deposit: convert USDC→XLM (backend), then mint
114 USDC_CONVERT_AND_MINT_DLQ: "usdc_convert_and_mint_dlq",
115 AUDIT_LOGS: "audit_logs",
116 AUDIT_LOGS_DLQ: "audit_logs_dlq",
117 } as const;
118
119 // Exchange names
120 export const EXCHANGES = {
121 RESERVE_EVENTS: "reserve_events",
122 TRANSACTION_EVENTS: "transaction_events",
123 DLX: "dlx", // Dead letter exchange
124 } as const;
Script executed:
# Check if mandatory or return options are used
rg "mandatory|return" src/services/audit/ --type typescriptRepository: Pi-Defi-world/acbu-backend
Repository: Pi-Defi-world/acbu-backend
Exit code: 2
stderr:
unrecognized file type: typescript
Script executed:
# Check if there are any uses of ConfirmChannel elsewhere in codebase
rg "ConfirmChannel|createConfirmChannel" --type tsRepository: Pi-Defi-world/acbu-backend
Repository: Pi-Defi-world/acbu-backend
Exit code: 1
Script executed:
# Check if callback or waitForConfirms is used anywhere
rg "waitForConfirms|sendToQueue.*function|sendToQueue.*callback" --type tsRepository: Pi-Defi-world/acbu-backend
Repository: Pi-Defi-world/acbu-backend
Exit code: 1
Use ConfirmChannel with publisher confirms for audit log delivery guarantees.
The sendToQueue() return value only reflects local write-buffer backpressure, not broker acknowledgement. true means "accepted into output buffer", false means "wait for drain"—neither confirms the broker received or persisted the message. This causes two problems:
- The code triggers fallback logging/alerts when merely experiencing backpressure (false positive), potentially duplicating audit records unnecessarily.
- The code treats a successful write-buffer accept (true) as broker delivery, missing messages lost on network failures or broker issues.
To ensure delivery: switch to ConfirmChannel via connection.createConfirmChannel(), use the callback signature sendToQueue(..., {}, (err, ok) => {...}) for per-message confirmation, and add mandatory: true with a 'return' event handler to detect unroutable publishes.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/services/audit/auditService.ts` around lines 30 - 45, The current use of
channel.sendToQueue(QUEUES.AUDIT_LOGS, ...) relies on the write-buffer boolean
and must be replaced with publisher confirms: create a ConfirmChannel via
connection.createConfirmChannel() and publish using ConfirmChannel.sendToQueue
with the callback form sendToQueue(..., { persistent: true, mandatory: true },
(err, ok) => { ... }) so you only treat the message as delivered on callback
success (err === null); also attach a channel 'return' event handler to detect
unroutable messages and handle/alert accordingly; update any logic in the audit
publishing path that currently throws on a boolean false to instead wait for the
confirm callback and handle err/return events.
| logger.error("CRITICAL: Audit logging failed (RabbitMQ unavailable)", { | ||
| eventType: entry.eventType, | ||
| error: error.message || error, | ||
| entry, | ||
| }); |
There was a problem hiding this comment.
Redact audit payloads before logging or emailing failures.
entry can carry arbitrary oldValue/newValue objects, and this path copies the full payload into general error logs and the alert email body. That widens access to potentially sensitive audit contents far beyond the audit store itself.
Also applies to: 83-90
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/services/audit/auditService.ts` around lines 59 - 63, The logger.error
call (and the alert email path) currently embeds the full audit entry object;
create and use a sanitizer (e.g. redactSensitiveFields or sanitizeAuditEntry)
that returns a copy of entry with sensitive keys removed or replaced (e.g. strip
or mask oldValue/newValue/any payload fields) and pass that sanitized object to
logger.error and to the email-sending routine (the alert email body path).
Update the code paths that reference entry (the logger.error call and the alert
email construction) to use the sanitized version instead of the raw entry to
avoid leaking sensitive audit payloads.
| try { | ||
| const logDir = path.dirname(config.logFile); | ||
| if (!fs.existsSync(logDir)) { | ||
| fs.mkdirSync(logDir, { recursive: true }); | ||
| } | ||
|
|
||
| const fallbackPath = path.join(logDir, "lost-audits.log"); | ||
| const fallbackEntry = { | ||
| timestamp: new Date().toISOString(), | ||
| ...entry, | ||
| error: error instanceof Error ? error.message : String(error), | ||
| }; | ||
|
|
||
| fs.appendFileSync(fallbackPath, JSON.stringify(fallbackEntry) + "\n"); | ||
| logger.info(`Audit entry saved to fallback file: ${fallbackPath}`); | ||
|
|
||
| // Alert admin | ||
| if (config.notification.alertEmail) { | ||
| const subject = `CRITICAL: Audit Log System Failure - ${entry.eventType}`; | ||
| const body = | ||
| `Audit logging failed to publish to RabbitMQ.\n\n` + | ||
| `Event Type: ${entry.eventType}\n` + | ||
| `Action: ${entry.action}\n` + | ||
| `Error: ${error.message || error}\n\n` + | ||
| `The audit entry has been saved to the fallback file: ${fallbackPath}\n\n` + | ||
| `Entry Data: ${JSON.stringify(entry, null, 2)}`; | ||
|
|
||
| sendEmail(config.notification.alertEmail, subject, body).catch((e) => { | ||
| logger.error("Failed to send audit failure alert email", { | ||
| error: e.message || e, | ||
| }); | ||
| }); | ||
| } | ||
| } catch (fallbackError: any) { |
There was a problem hiding this comment.
Still send the out-of-band alert when the file fallback fails.
sendEmail() is inside the same try as appendFileSync(), so disk/permission errors skip the only external alert and collapse the worst-case audit outage into another local log line. Move alerting outside the file-write try (or into a finally) so the “RabbitMQ down + fallback unavailable” case is still surfaced.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/services/audit/auditService.ts` around lines 65 - 98, The current try
block that writes the fallback file also contains the sendEmail call, so if
fs.appendFileSync or mkdirSync fails the alert email is skipped; refactor the
block in auditService.ts so that sendEmail(config.notification.alertEmail,
subject, body) is executed outside the filesystem try (or placed into a finally)
— keep building subject/body using entry and error as shown, wrap the sendEmail
call in its own catch to log failures via logger.error, and ensure
fallbackPath/logger messages remain in the file-write try while email alerting
is always attempted regardless of filesystem errors.
| describe('Simple', () => { | ||
| it('should pass', () => { | ||
| expect(1).toBe(1); | ||
| }); | ||
| }); |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Inconsistent with PR objectives: trivial test provides no coverage for the audit logging system.
The PR objectives describe implementing an asynchronous audit logging system with RabbitMQ, DLQ, retry logic, file fallback, and email alerts. However, this test file only contains a trivial assertion (expect(1).toBe(1)) that provides no actual test coverage for any of those features.
Consider adding meaningful tests that verify:
- Audit event publishing to RabbitMQ
- Retry logic for failed publishes
- Dead-letter queue handling
- File fallback persistence
- Critical alert notifications
🧰 Tools
🪛 ESLint
[error] 1-1: Replace 'Simple' with "Simple"
(prettier/prettier)
[error] 2-2: Replace 'should·pass' with "should·pass"
(prettier/prettier)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@tests/simple.test.ts` around lines 1 - 5, The current test only asserts a
trivial value and doesn't exercise the audit logging system; replace the trivial
spec with unit/integration tests that exercise the real code paths: call the
AuditPublisher.publishAuditEvent (or equivalent) to assert it enqueues to
RabbitMQClient.publish, simulate publish failures to verify the retry logic on
publishRetry, assert messages routed to DLQHandler.handleDeadLetter after
exceeded retries, verify FileFallback.persist (or fallbackToFile) is invoked
when RabbitMQ is unavailable, and mock sendCriticalAlert (or
AlertService.notifyCritical) to confirm alerting on persistent failures; use
spies/mocks for RabbitMQClient, DLQHandler, FileFallback, and AlertService to
assert behavior and add an integration-style test that runs the async flow
end-to-end with a test RabbitMQ or in-memory emulator.
| describe('Simple', () => { | ||
| it('should pass', () => { |
There was a problem hiding this comment.
Fix prettier formatting violations.
Replace single quotes with double quotes to match the project's prettier configuration.
🎨 Proposed fix
-describe('Simple', () => {
- it('should pass', () => {
+describe("Simple", () => {
+ it("should pass", () => {
expect(1).toBe(1);
});
});📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| describe('Simple', () => { | |
| it('should pass', () => { | |
| describe("Simple", () => { | |
| it("should pass", () => { | |
| expect(1).toBe(1); | |
| }); | |
| }); |
🧰 Tools
🪛 ESLint
[error] 1-1: Replace 'Simple' with "Simple"
(prettier/prettier)
[error] 2-2: Replace 'should·pass' with "should·pass"
(prettier/prettier)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@tests/simple.test.ts` around lines 1 - 2, The test file uses single quotes
which violate the project's Prettier rules; update string delimiters in the test
by replacing single quotes with double quotes in the describe and it calls
(e.g., the "describe" and "it" string literals) so the file uses double-quoted
strings consistently per the project's Prettier configuration.
|
@N-thnI cl failed |
|
closes #56 |
|
@N-thnI your code is still failing |
2 participants
closes #56
Summary by CodeRabbit
New Features
Tests